package com.example.test;

import com.example.constant.KafkaTopic;
import com.example.entity.Order;
import com.example.entity.User;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerOrderExample {


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        Producer<String, Order> producer = new KafkaProducer<>(props);
           int userId = 1;

            while (userId < 1000){
                int orderId = 1000000 + userId;
                Order order = new Order(orderId, orderId, userId);
                ProducerRecord<String, Order> record = new ProducerRecord<>(KafkaTopic.orderTopic, String.valueOf(orderId), order);
                try {
                    producer.send(record);
                    producer.flush();
                    System.out.println("Message sent successfully");
                    userId++;
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {


                }
            }

            producer.close();

    }

}